#include "mylib.h"

#if HAVE_NET_EPOLL_H



static S_NET_EPOLL_HANDLE stObject;

static void *net_epoll_worker(void *arg);

static unsigned long hash_count(void *buf)
{
	return *(unsigned long *)buf;
}

static int hash_compare(void *a, void *b)
{
	return memcmp(a, b, 4);
}


#define NET_EPOLL_DEAL_PTHREAD_NUM 5
#define NET_EPOLL_DEAL_QUEUE 20

int net_epoll_init(int max_pthread, int max_queue)
{
	memset(&stObject, 0, sizeof(S_NET_EPOLL_HANDLE));
	Pthread_rwlock_init(&stObject.lock, NULL);
    stObject.epoll_fd = Epoll_create(NET_EPOLL_EVENTS_MAX);
	tpool_init(&stObject.ptpool, max_pthread, max_queue, 0);
	stObject.running = NET_STAT_RUNNING;
	stObject.pstHash = hash_create(0, 100, hash_count, hash_compare, NULL, NULL);
	if (stObject.pstHash == NULL)
	{
		err_sys("hash_create is error");
	}
	Pthread_create(&stObject.epoll_tid, NULL, net_epoll_worker, NULL);
	return 0;
}



void setblock(int s, int is_block)
{
	int opts=fcntl(s, F_GETFL);
	if(opts < 0) return;

	if(is_block)
		opts = opts & ~O_NONBLOCK;
	else
		opts = opts | O_NONBLOCK;
	fcntl(s, F_SETFL, opts);
}

static void net_epoll_install_listen_calback(PS_TASK_NET pstTask, CB_EVENT pEvent_cb)
{
	struct epoll_event event;
	bzero(&event, sizeof(struct epoll_event));
	pstTask->deal_cb = pEvent_cb;
	if (pstTask->sock_type == NET_TASK_TYPE_UDP) 
		pstTask->event_flag = EPOLLIN|EPOLLET;
	else
		pstTask->event_flag = EPOLLIN|EPOLLET|EPOLLRDHUP;
	event.events = pstTask->event_flag;
	event.data.fd = pstTask->sockfd;
	Epoll_ctl(stObject.epoll_fd, EPOLL_CTL_ADD, pstTask->sockfd, &event);
	//Pthread_rwlock_wrlock(&stObject.lock);
	hash_insert_list(stObject.pstHash, &pstTask->sockfd, pstTask);
	//Pthread_rwlock_unlock(&stObject.lock);
	return;
}

static void net_epoll_add_send_event(PS_TASK_NET pstTask)
{
	struct epoll_event event;
	int op;
	
	if (pstTask->event_flag == 0)
	{
		op = EPOLL_CTL_ADD;
		if (pstTask->sock_type == NET_TASK_TYPE_UDP) 
			pstTask->event_flag = EPOLLET;
		else
			pstTask->event_flag = EPOLLET|EPOLLRDHUP;
	}
	else
		op = EPOLL_CTL_MOD;
	
	//pstTask->deal_cb = pEvent_cb;
	pstTask->event_flag |= EPOLLOUT;
	bzero(&event, sizeof(struct epoll_event));
	event.events = pstTask->event_flag;
	event.data.fd = pstTask->sockfd;
	Epoll_ctl(stObject.epoll_fd, op, pstTask->sockfd, &event);
	if (op == EPOLL_CTL_ADD)
	{
		//Pthread_rwlock_wrlock(&stObject.lock);
		hash_insert_list(stObject.pstHash, &pstTask->sockfd, pstTask);
		//Pthread_rwlock_unlock(&stObject.lock);
	}
}

static void net_epoll_remove_send_event(PS_TASK_NET pstTask)
{
	struct epoll_event event;
	bzero(&event, sizeof(struct epoll_event));
	//pstTask->out_event_cb = NULL;
	pstTask->event_flag &= ~EPOLLOUT;
	free(pstTask->send_buff);
	pstTask->send_buff = NULL;
	pstTask->send_len = 0;
	if ((pstTask->event_flag & ~(EPOLLET|EPOLLRDHUP)) == 0)
	{
		pstTask->event_flag = 0;
		Epoll_ctl(stObject.epoll_fd, EPOLL_CTL_DEL, pstTask->sockfd, &event);
		//Pthread_rwlock_wrlock(&stObject.lock);
		hash_del_entry(stObject.pstHash, &pstTask->sockfd);
		//Pthread_rwlock_unlock(&stObject.lock);
	}
	else
	{
		event.events = pstTask->event_flag;
		event.data.fd = pstTask->sockfd;
		Epoll_ctl(stObject.epoll_fd, EPOLL_CTL_MOD, pstTask->sockfd, &event);
	}
}



PS_TASK_NET net_epoll_add_task(int cmd, struct sockaddr *cmd_addr, CB_EVENT in_cb, int buff_size)
{
	PS_TASK_NET pstTask;
	int on = 1;
	int addr_len;
	int ret;
	
	if(stObject.running != NET_STAT_RUNNING)
	{
		err_msg("the net epoll module have not initialized yet");
		return NULL;
	}

	addr_len = sizeof(struct sockaddr);
	pstTask = Malloc(sizeof(S_TASK_NET));
	bzero(pstTask, sizeof(S_TASK_NET));
	switch(cmd)
	{
		case NET_TASK_CMD_ADD_LISTENING_TCP:
			if (cmd_addr==NULL || in_cb==NULL)
			{
				err_msg("net_epoll_add_listening para error");
				free(pstTask);
				return NULL;
			}
			pstTask->sock_type = NET_TASK_TYPE_LISTENING_TCP;
			pstTask->sockfd = Socket(AF_INET, SOCK_STREAM, 0);
			pstTask->mem_size = buff_size;
			setblock(pstTask->sockfd, SET_NONBLOCK);
			setsockopt(pstTask->sockfd, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on));
			Bind(pstTask->sockfd, cmd_addr, addr_len);
			Listen(pstTask->sockfd, 10);
			net_epoll_install_listen_calback(pstTask, in_cb);
			break;
		case NET_TASK_CMD_ADD_UDP:
			pstTask->sock_type = NET_TASK_TYPE_UDP;
			pstTask->sockfd = Socket(AF_INET, SOCK_DGRAM, 0);
			setblock(pstTask->sockfd, SET_NONBLOCK);
			if (cmd_addr != NULL)
				Bind(pstTask->sockfd, cmd_addr, addr_len);
			if (cmd_addr!=NULL && in_cb!=NULL)
			{
				pstTask->mem_size = buff_size;
				pstTask->recv_buf = Malloc(pstTask->mem_size);
				net_epoll_install_listen_calback(pstTask, in_cb);
			}
			break;
		case NET_TASK_CMD_ADD_CONNECT_TCP:
			if (cmd_addr == NULL)
			{
				err_msg("net_epoll_add_listening para error");
				free(pstTask);
				return NULL;
			}
			pstTask->sock_type = NET_TASK_TYPE_CONNECT_TCP;
			pstTask->sockfd = Socket(AF_INET, SOCK_STREAM, 0);
			setblock(pstTask->sockfd, SET_NONBLOCK);
			ret = connect_nonb(pstTask->sockfd, cmd_addr, addr_len, 500);
			if (ret == -1)
			{
				err_msg("net_epoll_add_task connect error");
				free(pstTask);
				return NULL;
			}
			if (in_cb != NULL)
			{
				pstTask->mem_size = buff_size;
				pstTask->recv_buf = Malloc(pstTask->mem_size);
				net_epoll_install_listen_calback(pstTask, in_cb);
			}
			break;
		default:
			err_msg("net_epoll_add_task illegal oper");
			free(pstTask);
			return NULL;
			break;
	}

	return pstTask;
}


void net_epoll_del_task(PS_TASK_NET pstTask)
{
	if (pstTask == NULL)
		return;
	
	if (pstTask->sockfd != -1)
	{
		if (pstTask->event_flag != 0)
		{
			Epoll_ctl(stObject.epoll_fd, EPOLL_CTL_DEL, pstTask->sockfd, NULL);
			//Pthread_rwlock_wrlock(&stObject.lock);
			hash_del_entry(stObject.pstHash, &pstTask->sockfd);
			//Pthread_rwlock_unlock(&stObject.lock);
		}
		close(pstTask->sockfd);
	}
	if (pstTask->recv_buf != NULL)
		free(pstTask->recv_buf);
	if (pstTask->send_buff != NULL)
		free(pstTask->send_buff);
	free(pstTask);
}

static void net_epoll_clean_task(PS_TASK_NET pstTask)
{
	if (pstTask->sockfd != -1)
	{
		if (pstTask->event_flag != 0)
		{
			Epoll_ctl(stObject.epoll_fd, EPOLL_CTL_DEL, pstTask->sockfd, NULL);
			//Pthread_rwlock_wrlock(&stObject.lock);
			hash_del_entry(stObject.pstHash, &pstTask->sockfd);
			//Pthread_rwlock_unlock(&stObject.lock);
		}
		close(pstTask->sockfd);
	}
	
	pstTask->event_flag = 0;
	pstTask->sockfd = -1;
}

static void do_send_data(void *pst)
{
	PS_TASK_NET pstTask;
	size_t		nleft;
	ssize_t		nwritten;
	const char	*ptr;
	int type;

	if (pst == NULL)
		return;
	pstTask = (PS_TASK_NET)pst;
	
	type = pstTask->sock_type;
	if (type == NET_TASK_TYPE_UDP || type == NET_TASK_TYPE_LISTENING_TCP)
	{
		err_msg("do_send_data err type");
		return;
	}
	ptr = pstTask->send_buff;
	nleft = pstTask->send_len;
	while (nleft > 0) 
	{
		if ( (nwritten = write(pstTask->sockfd, ptr, nleft)) <= 0) 
		{
			if (errno == EWOULDBLOCK)
				nwritten = 0;		/* and call write() again */
			else
			{
				
				pstTask->task_stat = NET_TASK_STAT_CLOSE;
				return;			/* error */
			}
		}

		nleft -= nwritten;
		ptr   += nwritten;
	}

	net_epoll_remove_send_event(pstTask);
	switch(type)
	{
		case NET_TASK_TYPE_CONNECT_TCP:			
			if (pstTask->task_stat == NET_TASK_STAT_CLOSE)
				net_epoll_clean_task(pstTask);		
			break;
		case NET_TASK_TYPE_CLIENT_TCP:
			if (pstTask->task_stat == NET_TASK_STAT_CLOSE)
				net_epoll_del_task(pstTask);	
			break;
		default:
			break;
	}
	
}

int net_epoll_send_date(PS_TASK_NET pstTask, struct sockaddr *dest_addr, const char *buff, size_t buff_len)
{
	int type;
	int salen;
	struct sockaddr sa;
	//int rlen;

	salen = sizeof(struct sockaddr);
	type = pstTask->sock_type;
	switch(type)
	{
		case NET_TASK_TYPE_LISTENING_TCP:
			err_msg("net_epoll_send_date socket type <NET_TASK_TYPE_LISTENING_TCP> cannot send date");
			return -1;
			break;
		case NET_TASK_TYPE_UDP:
			if (dest_addr == NULL)
				memcpy(&sa, &pstTask->recv_addr, sizeof(struct sockaddr));
			else
				memcpy(&sa, dest_addr, sizeof(struct sockaddr));
				
			if (sendto(pstTask->sockfd, buff, buff_len, 0, &sa, salen) != buff_len)
			{
				err_msg("net_epoll_send_date sendto error");
				return -1;
			}	
			break;
		case NET_TASK_TYPE_CONNECT_TCP:
		case NET_TASK_TYPE_CLIENT_TCP:
			if (pstTask->task_stat == NET_TASK_STAT_CLOSE)
			{
				err_msg("net_epoll_send_date the fd have been closed");
				return -1;
			}
			pstTask->send_len = buff_len;
			pstTask->send_buff = Malloc(buff_len);
			memcpy(pstTask->send_buff, buff, buff_len);
			net_epoll_add_send_event(pstTask);
			break;
	}

	return 0;
}

static void net_epoll_add_tcp_client(int client_fd, CB_EVENT in_cb, int buff_size)
{
	PS_TASK_NET pstTask;

	setblock(client_fd, SET_NONBLOCK);
	pstTask = Malloc(sizeof(S_TASK_NET));
	bzero(pstTask, sizeof(S_TASK_NET));
	pstTask->sockfd = client_fd;
	pstTask->sock_type = NET_TASK_TYPE_CLIENT_TCP;
	pstTask->mem_size = buff_size;
	pstTask->recv_buf = Malloc(pstTask->mem_size);
	net_epoll_install_listen_calback(pstTask, in_cb);
}

static void net_epoll_readn(PS_TASK_NET pstTask)
{
	size_t	nleft;
	ssize_t	nread;
	char	*ptr;
	
	ptr = pstTask->recv_buf;
	nleft = pstTask->mem_size;
	while (nleft > 0) 
	{
		if ( (nread = read(pstTask->sockfd, ptr, nleft)) < 0) 
		{
			if (errno != EWOULDBLOCK)
				err_sys("readn_unblock error");
			break;
		} 
		else if (nread == 0)
		{
			net_epoll_clean_task(pstTask);
			pstTask->task_stat = NET_TASK_STAT_CLOSE;
			break;				/* EOF */
		}
		else
		{
			nleft -= nread;
			ptr   += nread;
		}
	}
	
	pstTask->recv_len = pstTask->mem_size - nleft;
	
}

static void do_recv_data(void *pst)
{
	PS_TASK_NET pstTask;
	int type;
	//struct sockaddr sa;
	socklen_t socklen;

	
	if (pst == NULL)
		return;
		
	pstTask = (PS_TASK_NET)pst;

	type = pstTask->sock_type;

	switch(type)
	{
		case NET_TASK_TYPE_CONNECT_TCP:
			net_epoll_readn(pstTask);
			(*pstTask->deal_cb)(pstTask->recv_buf, pstTask->recv_len, (void *)pstTask);
			break;
		case NET_TASK_TYPE_CLIENT_TCP:
			net_epoll_readn(pstTask);
			(*pstTask->deal_cb)(pstTask->recv_buf, pstTask->recv_len, (void *)pstTask);
			if (pstTask->task_stat == NET_TASK_STAT_CLOSE)
				net_epoll_del_task(pstTask);
			break;
		case NET_TASK_TYPE_UDP:
			pstTask->recv_len = Recvfrom(pstTask->sockfd, pstTask->recv_buf, pstTask->mem_size, 0, &pstTask->recv_addr, &socklen);
			(*pstTask->deal_cb)(pstTask->recv_buf, pstTask->recv_len, (void *)pstTask);
			break;
		default:
			break;
	}
	return;
	
}


static void *net_epoll_worker(void *arg)
{
	int n;
	int nfds;
	int client_fd;
	struct sockaddr client_addr;
	socklen_t client_len;
	struct epoll_event events[NET_EPOLL_EVENTS_MAX];	
	PS_TASK_NET pstTask;
	sigset_t signal_set;
	
    sigfillset(&signal_set);
    pthread_sigmask(SIG_BLOCK, &signal_set, NULL);
    
    while(1)
    {
    	bzero(events, sizeof(struct epoll_event)*NET_EPOLL_EVENTS_MAX);
    	nfds = Epoll_wait(stObject.epoll_fd, events, NET_EPOLL_EVENTS_MAX, -1);
    	for (n=0; n<nfds; n++)
    	{
    		pstTask = (PS_TASK_NET)hash_search_list(stObject.pstHash, &events[n].data.fd);
    		if (pstTask == NULL)
    		{
    			err_msg("fd (%d) dont find task", events[n].data.fd);
    			break;
    		}

    		if (events[n].events&EPOLLIN)
    		{
				switch (pstTask->sock_type)
				{
					case NET_TASK_TYPE_LISTENING_TCP:
						client_fd = Accept(pstTask->sockfd, &client_addr, &client_len);
						net_epoll_add_tcp_client(client_fd, pstTask->deal_cb, pstTask->mem_size);
						break;
					case NET_TASK_TYPE_CONNECT_TCP:			
					case NET_TASK_TYPE_CLIENT_TCP:						
					case NET_TASK_TYPE_UDP:
						tpool_add_work(stObject.ptpool, do_recv_data, (void *)pstTask);
						break;
					default:
						break;
				}
			}
			else if (events[n].events&EPOLLOUT)
			{
				switch(pstTask->sock_type)
				{
					case NET_TASK_TYPE_CONNECT_TCP:
					case NET_TASK_TYPE_CLIENT_TCP:
						tpool_add_work(stObject.ptpool, do_send_data, (void *)pstTask);
						break;
					default:
						break;
				}
			}
			else
			{
				switch(pstTask->sock_type)
				{
					case NET_TASK_TYPE_CONNECT_TCP:
						net_epoll_clean_task(pstTask);
						pstTask->task_stat = NET_TASK_STAT_CLOSE;
						break;
					case NET_TASK_TYPE_CLIENT_TCP:
						net_epoll_del_task(pstTask);
						break;
					default:
						break;
				}
			}
    	}
    }

    return 0;
}

int net_epoll_connect_is_health(PS_TASK_NET pstTask)
{
	if (pstTask == NULL)
		return 0;
	if (pstTask->sock_type == NET_TASK_TYPE_UDP)
		return 1;
	if (pstTask->task_stat == NET_TASK_STAT_CLOSE)
		return 0;
	else
		return 1;
}

#endif


